package com.youxin.dataStream.checkPoint;

import com.youxin.dataStream.custormSource.SimgleParalleSource;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;

/**
 * checkPoint:
 *  应用定时触发，用于保存状态，会过期
 *  内部应用失败重启的时候使用
 *
 *  savePoint
 *  用户手动执行，是指向checkpoint的指针，不会过期
 *  在升级的情况下使用
 *  注意：为了能够在作业的不同版本之间以及Flink的不同版本之间顺利升级，强烈推荐使用uid(String)方法手动的赋予算子赋予ID,
 *  这些ID将用于确定每一个算子的状态范围。如果不手动给个算子指定ID，否则由Flink自动给内个孙子生成一个ID。只要这些ID没有改变就能从保存点（savePoint）将程序恢复过来。
 *  二这些自动生成的ID依赖与程序的结构，并且代码的更改是很敏感的。因此强烈建议用手动的设置ID
 *
 *  在程序执行的时候可以进行savepoint   也可以执行停任务的时候进行savepoint
 *
 *
 *  可以在cancel
 *
 *
 *  savepoint位置
 *  state.savepoint.dir: hdfs://
 *
 */
public class SavePointText {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 每隔1000 ms进行启动一个检查点【设置checkpoint的周期】
        env.enableCheckpointing(1000);
// 高级选项：
// 设置模式为exactly-once （这是默认值）
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
// 确保检查点之间有至少500 ms的间隔【checkpoint最小间隔】
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);
// 检查点必须在一分钟内完成，或者被丢弃【checkpoint的超时时间】
        env.getCheckpointConfig().setCheckpointTimeout(60000);
// 同一时间只允许进行一个检查点
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
// 表示一旦Flink处理程序被cancel后，会保留Checkpoint数据，以便根据实际需要恢复到指定的Checkpoint【详细解释见备注】
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);



        DataStreamSource<Long> text = env.addSource(new SimgleParalleSource()).setParallelism(1);//并行度只能为1   Source: 1 is not a parallel source
        text.print();
        SingleOutputStreamOperator<Long> sum = text.timeWindowAll(Time.seconds(2)).sum(0)
                .uid("sum-id");//添加id，不指定的话其实会自动生成id
        sum.print().setParallelism(1);
        env.execute("simgleParalle");
    }
}
